package com.xgq.drink.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.xgq.drink.config.KafkaConfig;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.xgq.drink.entity.ChatMessage;
import com.xgq.drink.entity.R;
import com.xgq.drink.mapper.ChatAttachmentMapper;
import com.xgq.drink.mapper.ChatMessageMapper;
import com.xgq.drink.service.IChatMessageService;
import com.xgq.drink.utils.RUtils;
import com.xgq.drink.utils.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.kafka.annotation.KafkaListener;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Properties;

/**
 * 聊天消息服务实现类
 */
@Service
@Slf4j
public class ChatMessageServiceImpl extends ServiceImpl<ChatMessageMapper, ChatMessage> implements IChatMessageService {
    @Autowired
    private ChatMessageMapper chatMessageMapper;

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Autowired
    private ChatAttachmentMapper chatAttachmentMapper;

    /**
     * 发送私聊消息
     */
    public R sendPrivateMessage(String senderId, String receiverId, Object content, Integer messageType) {
        log.info("执行私聊！！！！！！！！！！！！1");
        try {
            ChatMessage chatMessage = new ChatMessage();
            chatMessage.setSenderId(Long.valueOf(senderId));
            chatMessage.setReceiverId(Long.valueOf(receiverId));
            chatMessage.setChatType(1); // 1表示私聊
            chatMessage.setStatus(0); // 0表示未读
            chatMessage.setSendStamp(TimeUtils.getCurrentTimestamp());
            boolean saved = this.save(chatMessage);
            if (saved) {
                kafkaTemplate.send("private-chat", chatMessage);
                return RUtils.success("消息发送成功", chatMessage);
            } else {
                return RUtils.fail("消息发送失败");
            }
        } catch (Exception e) {
            log.error("发送私聊消息异常", e);
            return RUtils.fail("发送消息异常: " + e.getMessage());
        }
    }

    /**
     * 发送群聊消息
     */
    public R sendGroupMessage(String senderId, String groupId, Object content, Integer messageType) {
        log.info("执行群聊！！！！！！！！！！！！1");
        try {
            ChatMessage chatMessage = new ChatMessage();
            chatMessage.setSenderId(Long.valueOf(senderId));
            chatMessage.setReceiverId(Long.valueOf(groupId));
            chatMessage.setChatType(2);
            chatMessage.setStatus(0);
            chatMessage.setSendStamp(TimeUtils.getCurrentTimestamp());
            boolean saved = this.save(chatMessage);
            if (saved) {
                // 发送到Kafka消息队列
                kafkaTemplate.send("group-chat", chatMessage);
                return RUtils.success("消息发送成功", chatMessage);
            } else {
                return RUtils.fail("消息发送失败");
            }
        } catch (Exception e) {
            log.error("发送群聊消息异常", e);
            return RUtils.fail("发送消息异常: " + e.getMessage());
        }
    }

    /**
     * 标记消息为已读
     */
    @Override
    public R markMessageAsRead(String messageId) {
        try {
            if (messageId == null) {
                return RUtils.fail("消息ID不能为空");
            }

            int result = chatMessageMapper.markMessageAsRead(Long.valueOf(messageId));
            if (result > 0) {
                return RUtils.success("标记已读成功");
            } else {
                return RUtils.fail("标记已读失败");
            }
        } catch (Exception e) {
            log.error("标记消息已读异常", e);
            return RUtils.fail("标记已读异常: " + e.getMessage());
        }
    }

    /**
     * 标记用户所有未读消息为已读
     */
    @Override
    public R markAllMessagesAsRead(String receiverId, String senderId) {
        try {
            if (receiverId == null || senderId == null) {
                return RUtils.fail("参数不能为空");
            }
            int result = chatMessageMapper.markAllMessagesAsRead(Long.valueOf(receiverId), Long.valueOf(senderId));
            return RUtils.success("标记全部已读成功", result);
        } catch (Exception e) {
            log.error("标记全部已读异常", e);
            return RUtils.fail("标记全部已读异常: " + e.getMessage());
        }
    }

    @Override
    public List<Map<String, Object>> fetchPaginatedMessages(String receiverId, String senderId, int page,
            int pageSize) {
        System.out.println("查！");
        // 计算分页的起始位置
        int offset = (page - 1) * pageSize;

        System.out.println("senderId:" + senderId);
        // 查询数据库中的消息记录
        List<ChatMessage> messages = chatMessageMapper.getPrivateChatHistory(receiverId, senderId,offset, pageSize);

        // 将结果转换为 Map 格式
        return messages.stream().map(msg -> {
            Map<String, Object> messageMap = new HashMap<>();
            messageMap.put("messageId", msg.getMessageId());
            messageMap.put("senderId", msg.getSenderId());
            messageMap.put("receiverId", msg.getReceiverId());
            messageMap.put("content", msg.getContent());
            messageMap.put("chatType", msg.getChatType());
            messageMap.put("messageType", msg.getMessageType());
            messageMap.put("status", msg.getStatus());
            messageMap.put("timestamp", msg.getSendStamp());
            if(msg.getMessageType() != 1) {
                QueryWrapper wrapper = new QueryWrapper();
                wrapper.eq("message_id", msg.getMessageId());
                List<Map<String, Object>> attachments = chatAttachmentMapper.selectList(wrapper);
                    messageMap.put("attachments", attachments);
            }
            return messageMap;
        }).collect(Collectors.toList());
    }

    @Override
    public List<Map<String, Object>> fetchGroupMessages(String groupId, int page, int pageSize) {
        // 计算分页的起始位置
        int offset = (page - 1) * pageSize;

        // 查询数据库中的群聊消息记录
        List<ChatMessage> messages = chatMessageMapper.getGroupChatHistory(groupId, pageSize, offset);

        // 将结果转换为 Map 格式
        return messages.stream().map(msg -> {
            Map<String, Object> messageMap = new HashMap<>();
            messageMap.put("messageId", msg.getMessageId());
            messageMap.put("senderId", msg.getSenderId());
            messageMap.put("groupId", msg.getGroupId());
            messageMap.put("content", msg.getContent());
            messageMap.put("chatType", msg.getChatType());
            messageMap.put("status", msg.getStatus());
            messageMap.put("timestamp", msg.getSendStamp());
            return messageMap;
        }).collect(Collectors.toList());
    }
}
